package com.rxjava;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.schedulers.Schedulers;

//RxJava中还有许多操作符。操作符就是用于在Observable和最终的Observer之间，通过转换Observable为其他观察者对象的过程，修改发出的事件，最终将最简洁的数据传递给Observer对象。
public class OperatorDemo {

	public static void main(String[] args) {
		//map()操作符，就是把原来的Observable对象转换成另一个Observable对象，同时将传输的数据进行一些灵活的操作，方便Observer获得想要的数据形式。
		Observable.just("hello").map(s -> s.length()).subscribe(len -> System.out.println(len));

		List<String> list = new ArrayList<>();
		list.add("from1");
		list.add("from2");
		list.add("from3");
		//flatMap()对于数据的转换比map()更加彻底，如果发送的数据是集合，flatmap()重新生成一个Observable对象，并把数据转换成Observer想要的数据形式。它可以返回任何它想返回的Observable对象。
		Observable.just(list).flatMap(strings -> Observable.fromIterable(strings))
		//take()操作符：输出最多指定数量的结果。
		.take(2)
		//doOnNext()允许我们在每次输出一个元素之前做一些额外的事情。
		.doOnNext(str -> System.out.println("准备工作"))
		.subscribe(str -> System.out.println(str));

		List<String> list2 = new ArrayList<>();
		list2.add("zsl");
		list2.add("chenli");
		list2.add("wujian122");
		Observable.just(list2).flatMap(strings -> Observable.fromIterable(strings))
		.filter(str -> str.length() > 4)
		.subscribe(str -> System.out.println(str));

		//分组
		Observable.just(1, 2, 3, 4, 5).groupBy(num -> num % 2)
		.subscribe(new Observer<GroupedObservable<Integer, Integer>>() {

			@Override
			public void onComplete() {

			}

			@Override
			public void onError(Throwable arg0) {

			}

			@Override
			public void onNext(GroupedObservable<Integer, Integer> arg0) {
				arg0.subscribe(new Observer<Integer>() {

					@Override
					public void onComplete() {

					}

					@Override
					public void onError(Throwable arg0) {

					}

					@Override
					public void onNext(Integer data) {
						System.out.println("group:" +arg0.getKey() + ", data:" + data);
					}

					@Override
					public void onSubscribe(Disposable arg0) {

					}



				});
			}

			@Override
			public void onSubscribe(Disposable arg0) {

			}

		});

		//buffer：把单个数据源聚合起来一起发送
		Observable.just(1, 2, 3, 4, 5).buffer(2).subscribe(xlist -> System.out.println(xlist));

		//scan：把前两个值累加，和再与第三个值累加，以此类推
		Observable.just(1, 2, 3, 4, 5).scan((sum, i) -> (sum + i)).subscribe(sum -> System.out.println(sum));

		//window操作符会在时间间隔内缓存结果，类似于buffer缓存一个list集合，区别在于window将这个结果集合封装成了observable
		//window(long timespan, TimeUnit unit)
		//第一个是缓存的间隔时间，第二个参数是时间单位
		Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline()).take(10).window(3, TimeUnit.SECONDS).subscribe(new Observer<Observable<Long>>() {
            @Override
            public void onComplete() {
                System.out.println("------>onCompleted()");
            }

            @Override
            public void onError(Throwable e) {
            	System.out.println("------>onError()" + e);
            }

            @Override
            public void onNext(Observable<Long> integerObservable) {
            	System.out.println("------->onNext()");
                integerObservable.subscribe(integer -> System.out.println("------>call():" + integer));
            }

			@Override
			public void onSubscribe(Disposable arg0) {

			}

        });

		//在Eclipse下测试Rxjava中的操作符interval()时出现了很奇怪的问题，怎么试都不能执行。
		//原因是我们的操作不是阻塞的：我们创建了一个每隔一段时间就发射数据的 Observable，然后我们注册了一个 Subscriber 来打印收到的数据。这两个操作都是非阻塞的，而 发射数据的计时器是运行在另外一个线程的，但是这个线程不会阻止 JVM 结束当前的程序，所以 如果没有 System.in.read(); 这个阻塞操作，还没发射数据则程序就已经结束运行了。
		//Observable.interval(1, TimeUnit.SECONDS).subscribe(aLong -> System.out.println("------ onNext, aLong : " + aLong));
		//Schedulers.trampoline()主要用于延迟工作任务的执行。当我们想在当前线程执行一个任务时，并不是立即，我们可以用.trampoline()将它入队。
	    //trampoline将会处理它的队列并且按序运行队列中每一个任务。它是repeat()和retry()方法默认的调度器。
		//Observable.interval(1, TimeUnit.SECONDS, Schedulers.trampoline()).subscribe(aLong -> System.out.println("------ onNext, aLong : " + aLong));

	}

}
